package parallel;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.InputFormat;
import org.apache.hadoop.mapreduce.InputSplit;
import org.apache.hadoop.mapreduce.JobContext;
import org.apache.hadoop.mapreduce.RecordReader;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import event.Basic;
import event.BasicConfig;

public class BasicRangeInputFormat extends InputFormat<LongWritable, NullWritable> {

	/**
	 * An input split consisting of a range on numbers.
	 */

	/**
	 * A record reader that will generate a range of numbers.
	 */
	@Override
	public RecordReader<LongWritable, NullWritable> createRecordReader(
			InputSplit split,
			TaskAttemptContext context) throws IOException,
			InterruptedException {
		// TODO Auto-generated method stub
		return new BasicRangeRecordReader((BasicRangeInputSplit)split);
	}

	@Override
	public List<InputSplit> getSplits(
			JobContext context) throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		int numSplits = getNumberOfRegions(context);
		BasicConfig config = new BasicConfig();
		String type = getType(context);
		Basic basic = config.getBasic(type);
		long totalsequences = basic.getNumber();
		long sequencesPerSplit = totalsequences / numSplits;
		System.out.println("Generating " + totalsequences + " using " + numSplits + 
				" maps with step of " + sequencesPerSplit);
		List<InputSplit> splits = new ArrayList<InputSplit>();
		long currentsequence = 0;
		for(int split = 0; split < numSplits-1; ++split) {
			BasicRangeInputSplit inputSplit = new BasicRangeInputSplit(sequencesPerSplit, split);
			splits.add(inputSplit);
			currentsequence += sequencesPerSplit;
		}
		System.out.println("current sequence: " + sequencesPerSplit);
		long sequencesLeft = totalsequences - currentsequence;
		BasicRangeInputSplit lastSplit = new BasicRangeInputSplit(sequencesLeft, numSplits-1);
		splits.add(lastSplit);
		for(int i = 0; i < splits.size(); i++)
		{
			System.out.println("splits id: "+ ((BasicRangeInputSplit)splits.get(i)).getSplitId());
		}
		return splits;

	}
	int getNumberOfRegions(JobContext context) {
		Configuration conf = context.getConfiguration();
		return conf.getInt("mapred.map.tasks", 1);
	}
	String getType(JobContext context) {
		Configuration conf = context.getConfiguration();
		String sequences = conf.get("Basic-type", null);
		return sequences;
	}

}
